package org.raymond.icloud.common.starter.autoconfiguration;

import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.raymond.icloud.common.core.constants.MQConstants;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.batch.SimpleBatchingStrategy;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.DependsOn;
import org.springframework.retry.RetryPolicy;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.policy.CompositeRetryPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.policy.TimeoutRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;

import java.text.SimpleDateFormat;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;

@Slf4j
@EnableRabbit
@Configuration
public class RabbitMQConfiguration {

    @Bean("logDlxQueue")
    public Queue logDlxQueue(){
        return QueueBuilder.durable(MQConstants.QUEUEE_LOG_DLX).build();
    }

    @Bean("logDlxDirectExchange")
    public DirectExchange logDlxDirectExchange(){
        return (DirectExchange) ExchangeBuilder.directExchange(MQConstants.EXCHANGE_LOG_DLX).durable(true).build();
    }

    @Bean
    public Binding logDlxBinding(@Qualifier("logDlxQueue") Queue queue, @Qualifier("logDlxDirectExchange") DirectExchange directExchange){
        return BindingBuilder.bind(queue).to(directExchange).with(MQConstants.ROUTING_LOG_DLX);
    }

    @Bean("logQueue")
    public Queue logQueue(){
        Map<String, Object> map = new HashMap<>();
        map.put("x-dead-letter-exchange", MQConstants.EXCHANGE_LOG_DLX);
        map.put("x-dead-letter-routing-key", MQConstants.ROUTING_LOG_DLX);
        map.put("x-message-ttl", 10*1000);
        return QueueBuilder.durable(MQConstants.QUEUEE_LOG_COMMON).withArguments(map).build();
    }

    @Bean("logDirectExchange")
    public DirectExchange logDirectExchange(){
        return (DirectExchange) ExchangeBuilder.directExchange(MQConstants.EXCHANGE_LOG_COMMON).durable(true).build();
    }

    @Bean
    public Binding logBinding(@Qualifier("logQueue") Queue queue, @Qualifier("logDirectExchange") DirectExchange directExchange){
        return BindingBuilder.bind(queue).to(directExchange).with(MQConstants.ROUTING_LOG);
    }

    @Bean("userQueue")
    public Queue userQueue(){
        return QueueBuilder.durable(MQConstants.QUEUEE_USER_COMMON).build();
    }

    @Bean("userDirectExchange")
    public DirectExchange userDirectExchange(){
        return (DirectExchange) ExchangeBuilder.directExchange(MQConstants.EXCHANGE_USER_COMMON).durable(true).build();
    }

    @Bean
    public Binding userBinding(@Qualifier("userQueue") Queue queue, @Qualifier("userDirectExchange") DirectExchange directExchange){
        return BindingBuilder.bind(queue).to(directExchange).with(MQConstants.ROUTING_USER);
    }

    @Bean
    public Jackson2JsonMessageConverter jsonMessageConverter(){
        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.setDateFormat(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS"));
        objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES,false);
        //objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
        return new Jackson2JsonMessageConverter(objectMapper);
    }

    @Bean
    public CachingConnectionFactory producerConnectionFactory(){
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setUsername("icloud");
        factory.setPassword("icloud123");
        factory.setVirtualHost("/icloud");

        factory.setConnectionNameStrategy(connectionFactory -> "mq-producer-conn-");
        /**
         * 连接会被缓存起来,createConnection()的调用可能会创建一个新的连接,也可能从缓存中获取;
         * 关闭连接时,如果缓存的连接数没有达到cache size,就会将连接缓存起来.在这个模式下,由连接创建的channel也会被缓存起来
         * 在高可用集群下,由于连接连着不同的broker,缓存连接会变得非常有用
         * 在CacheMode.CONNECTION模式下,不支持amqp中各个角色的automatic declaration
         */
        //factory.setCacheMode(CachingConnectionFactory.CacheMode.CONNECTION);
        /**
         * 当设置为CacheMode.CONNECTION,spring amqp提供了一个属性connectionLimit,要配合channelCheckoutTimeLimit使用.
         * 注意,connectionLimit不代表连接的限制数,当连接数达到connectionLimit时,spring amqp会监视存活时间超过channelCheckoutTimeLimit的连接,
         * 当连接超过这个时间时,会抛出AmqpTimeoutException.因此,connectionLimit在一定意义上指的是空闲连接数
         */
        factory.setConnectionLimit(Integer.MAX_VALUE);

        factory.setChannelCheckoutTimeout(60*1000);
        /**
         * the amqp-client library by default creates a fixed thread pool for each connection (default size: Runtime.getRuntime().availableProcessors() * 2 threads).
         * When using a large number of connections, you should consider setting a custom executor on the CachingConnectionFactory.
         * Then, the same executor can be used by all connections and its threads can be shared.
         * The executor’s thread pool should be unbounded or set appropriately for the expected use (usually, at least one thread per connection).
         * If multiple channels are created on each connection, the pool size affects the concurrency, so a variable (or simple cached) thread pool executor would be most suitable.
         */
        factory.setExecutor(Executors.newCachedThreadPool());

        /**
         * by default, establishes a single connection proxy that can be shared by the application.
         */
        factory.setCacheMode(CachingConnectionFactory.CacheMode.CHANNEL);
        /**
         * 当channelCheckoutTimeout大于0时,channelCacheSize代表一个连接可以创建的最大channel数,当达到channelCacheSize时,新的chanel申请会被阻塞,
         * 如果阻塞超时会抛出AmqpTimeoutException
         * 当在框架中使用RabbitTemplate时,Channels可以得到有效的管理.如果通过createChannel直接创建channel,需要通过手动关闭channel(调用closing)
         */
        /**
         * Because of the async nature of RabbitMQ and the use of cached channels;
         * it is not certain that the same channel will be used and therefore the order in which the messages arrive in the queue is not guaranteed.
         * (In most cases they will arrive in order, but the probability of out-of-order delivery is not zero).
         * To solve this use case, you can use a bounded channel cache with size 1 (together with a channelCheckoutTimeout)
         * to ensure the messages are always published on the same channel, and order will be guaranteed.
         * To do this, if you have other uses for the connection factory, such as consumers, you should either use a dedicated connection factory
         * for the template, or configure the template to use the publisher connection factory embedded in the main connection factory
         * publisherCF.setChannelCacheSize(1);
         * publisherCF.setChannelCheckoutTimeout(1000L);
         */
        factory.setChannelCacheSize(25);

        factory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        factory.setPublisherReturns(true);

        factory.setConnectionTimeout(15*1000);
        factory.setRequestedHeartBeat(60);

        return factory;
    }

    @Bean
    public CachingConnectionFactory consumerConnectionFactory(){
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setUsername("icloud");
        factory.setPassword("icloud123");
        factory.setVirtualHost("/icloud");

        factory.setConnectionNameStrategy(connectionFactory -> "mq-consumer-conn-");
        //factory.setCacheMode(CachingConnectionFactory.CacheMode.CONNECTION);
        factory.setConnectionLimit(Integer.MAX_VALUE);
        factory.setChannelCheckoutTimeout(60*1000);
        factory.setExecutor(Executors.newCachedThreadPool());
        factory.setCacheMode(CachingConnectionFactory.CacheMode.CHANNEL);
        factory.setChannelCacheSize(25);
        factory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        factory.setPublisherReturns(true);
        factory.setConnectionTimeout(15*1000);
        factory.setRequestedHeartBeat(60);
        return factory;
    }

    private RetryTemplate retryTemplate(){
        RetryTemplate retryTemplate = new RetryTemplate();
        ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
        backOffPolicy.setInitialInterval(5*1000);
        backOffPolicy.setMaxInterval(10*1000);
        backOffPolicy.setMultiplier(1.5);
        retryTemplate.setBackOffPolicy(backOffPolicy);

        SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy(5);
        TimeoutRetryPolicy timeoutRetryPolicy = new TimeoutRetryPolicy();
        timeoutRetryPolicy.setTimeout(5*1000);
        RetryPolicy[] retryPolicies = new RetryPolicy[]{simpleRetryPolicy, timeoutRetryPolicy};
        CompositeRetryPolicy compositeRetryPolicy = new CompositeRetryPolicy();
        compositeRetryPolicy.setPolicies(retryPolicies);

        retryTemplate.setRetryPolicy(simpleRetryPolicy);
        return retryTemplate;
    }

    @Bean
    @DependsOn(value={"producerConnectionFactory", "jsonMessageConverter"})
    public RabbitTemplate rabbitTemplate(@Qualifier("producerConnectionFactory") CachingConnectionFactory connectionFactory,
        @Qualifier("jsonMessageConverter") Jackson2JsonMessageConverter jsonMessageConverter){
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setRetryTemplate(retryTemplate());

        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) ->
                log.debug("message to exchange-correlationData:{}, ack:{}, cause:{}", correlationData, ack, cause));
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingkey) -> {
            String correlationId = message.getMessageProperties().getCorrelationId();
            String messageId = message.getMessageProperties().getMessageId();
            log.debug("messageId:{}, correlationId:{}, exchange:{}, routingKey:{}, replyCode:{}, replyText:{}",
                    messageId, correlationId, exchange, routingkey, replyCode, replyText);
        });

        rabbitTemplate.setMessageConverter(jsonMessageConverter);
        return rabbitTemplate;
    }

    @Bean
    @DependsOn(value={"producerConnectionFactory", "jsonMessageConverter"})
    public BatchingRabbitTemplate batchingRabbitTemplate(@Qualifier("producerConnectionFactory") CachingConnectionFactory connectionFactory,
                                                         @Qualifier("jsonMessageConverter") Jackson2JsonMessageConverter jsonMessageConverter){
        SimpleBatchingStrategy batchingStrategy = new SimpleBatchingStrategy(10, 1024*1024, 5*1000);
        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
        scheduler.setPoolSize(1);
        scheduler.setDaemon(true);
        scheduler.setThreadNamePrefix("mq-batching-");
        scheduler.setAwaitTerminationSeconds(60);
        scheduler.setWaitForTasksToCompleteOnShutdown(true);
        BatchingRabbitTemplate batchTemplate = new BatchingRabbitTemplate(connectionFactory, batchingStrategy,scheduler);

        batchTemplate.setRetryTemplate(retryTemplate());

        batchTemplate.setMandatory(true);
        batchTemplate.setConfirmCallback((correlationData, ack, cause) ->
                log.debug("message to exchange-correlationData:{}, ack:{}, cause:{}", correlationData, ack, cause));
        batchTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingkey) -> {
            String correlationId = message.getMessageProperties().getCorrelationId();
            String messageId = message.getMessageProperties().getMessageId();
            log.debug("messageId:{}, correlationId:{}, exchange:{}, routingKey:{}, replyCode:{}, replyText:{}",
                    messageId, correlationId, exchange, routingkey, replyCode, replyText);
        });

        batchTemplate.setMessageConverter(jsonMessageConverter);
        return batchTemplate;
    }

    @Bean
    @DependsOn(value={"consumerConnectionFactory", "jsonMessageConverter"})
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
            @Qualifier("consumerConnectionFactory") CachingConnectionFactory connectionFactory,
            @Qualifier("jsonMessageConverter") Jackson2JsonMessageConverter jsonMessageConverter){

        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(jsonMessageConverter);
        factory.setDefaultRequeueRejected(false);
        factory.setAutoStartup(true);

        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        factory.setReceiveTimeout(10*1000L);
        factory.setPrefetchCount(10);
        factory.setConcurrentConsumers(1);
        factory.setMaxConcurrentConsumers(1);
        factory.setTaskExecutor(Executors.newSingleThreadExecutor());
        factory.setBatchListener(true);
        factory.setBatchSize(10);
        factory.setConsumerBatchEnabled(true);
        factory.setDeBatchingEnabled(true);
        /**
         * container is idle: some time passes with no message delivery
         */
        factory.setIdleEventInterval(60*1000L);

        factory.setBeforeSendReplyPostProcessors(message->{
            log.debug("before reply: {}", message);
            return message;
        });
        factory.setAfterReceivePostProcessors(message -> {
            log.debug("after reply: {}", message);
            return message;
        });
        factory.setRetryTemplate(retryTemplate());

        factory.setErrorHandler(t-> log.error("Listened error:", t));
        return factory;
    }

    @Bean
    @DependsOn(value={"rabbitTemplate"})
    public RabbitAdmin rabbitAdmin(RabbitTemplate rabbitTemplate){
        RabbitAdmin rabbitAdmin = new RabbitAdmin(rabbitTemplate);
        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }

    /*
    public ConditionalRejectingErrorHandler conditionalRejectingErrorHandler(){
        ConditionalRejectingErrorHandler conditionalRejectingErrorHandler = new ConditionalRejectingErrorHandler();
        conditionalRejectingErrorHandler.setRejectManual(false);
        return conditionalRejectingErrorHandler;
    }
     */
}
